package com.haishuibutian.rocketmqconsumer.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

@Service
public class MQConsumer {
    private static final Logger logger = LoggerFactory.getLogger(MQConsumer.class);
    private DefaultMQPushConsumer consumer;

    @Autowired
    private MQConfig mqConfig;
    @Autowired
    private MQListener mqListener;

    @PostConstruct
    public void init() throws MQClientException {
        logger.info("MQ消费者初始化...");
        consumer = new DefaultMQPushConsumer(mqConfig.getGroupName());
        consumer.setNamesrvAddr(mqConfig.getNameSrv());
        consumer.subscribe(mqConfig.getProducerTopic(), "*");
        if (MessageModel.CLUSTERING.getModeCN().equals(mqConfig.getMessageModel())) {
            consumer.setMessageModel(MessageModel.CLUSTERING);
        } else {
            consumer.setMessageModel(MessageModel.BROADCASTING);
        }
        consumer.setConsumeMessageBatchMaxSize(mqConfig.getBatchMaxSize());
        consumer.setConsumeThreadMin(mqConfig.getConsumeThreadMin());
        consumer.setConsumeThreadMax(mqConfig.getConsumeThreadMax());
        consumer.registerMessageListener(mqListener);
        consumer.start();
        logger.info("MQ消费者启动成功 -> nameSrv={},groupName={},topic={},messageModel={},consumeThreadMin={},consumeThreadMax={}"
                , mqConfig.getNameSrv(), mqConfig.getGroupName(), mqConfig.getProducerTopic(), mqConfig.getMessageModel(), mqConfig.getConsumeThreadMin(), mqConfig.getConsumeThreadMax());
    }

    @PreDestroy
    public void destroy() {
        if (consumer != null) {
            consumer.shutdown();
            logger.info("MQ消费者销毁...");
        }
    }
}
